package com.atguigu.Flink.wordCount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//一个作业需要用多少个slot?
//        *    一个作业的并行度是多少就需要用多少个slot.
//        *
//        * 任务槽共享:
//        *    Flink允许多个task共享同一个Slot 。
//        *
//        * 哪些task可以共享同一个Slot？
//        *    不同任务节点(算子)的上下游task可以共享一个Slot .
//        *
//        *
//        * Slot共享的好处:
//        *   1. 当我们将资源密集型和非密集型的任务同时放到一个slot中，它们就可以自行分配对资源占用的比例，从而保证最重的活平均分配给所有的TaskManager。
//        *   2. slot共享另一个好处就是允许我们保存完整的作业管道。这样一来，即使某个TaskManager出现故障宕机，其他节点也可以完全不受影响，作业的任务可以继续执行
//        *
//        * Slot的操作:
//        *    Flink默认情况下开启了Slot共享。

public class Flink08_Slot {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port" , 6789);
        // 1. 准备流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //设置全局并行度
        env.setParallelism(1) ;

        // 2. 从数据源读取数据
        // DataStreamSource => DataStream
        SingleOutputStreamOperator<String> ds =
                env.socketTextStream("hadoop102", 8888).slotSharingGroup("group1");

        // 3. 转换处理
        // 3.1 将读取到的一行数据按照分隔符进行切分，处理成（word, 1）的格式
        SingleOutputStreamOperator<Tuple2<String, Long>> flatMap = ds.flatMap(
                new FlatMapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                        String[] words = value.split(" ");
                        for (String word : words) {
                            out.collect(Tuple2.of(word, 1L));
                        }
                    }
                }
        ).slotSharingGroup("group2");


        //3.2 根据单词分组
        //KeyedStream => DataStream
        KeyedStream<Tuple2<String, Long>, String> keyBy = flatMap.keyBy(
                new KeySelector<Tuple2<String, Long>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Long> value) throws Exception {
                        return value.f0; // 使用Tuple的第一个元素作为Key
                    }
                }
        );
        //3.3 统计每个单词出现的次数
        /*
           sum(int) : 如果流中的数据类型是Tuple, 指定Tuple中的第几个元素用于sum汇总。
           sum(String) : 如果流中的数据类型是POJO ， 指定POJO中的哪个属性用于sum汇总。
         */
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = keyBy.sum(1).setParallelism(1) ;

        //4. 输出结果
        sum.print().setParallelism(1) ;

        //5. 启动执行
        env.execute();

    }
}
